package org.pangu.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author zhangyj
 * @date 2021/8/28 16:41
 */
@Configuration
@Slf4j
public class MQConsumerConfiguration {
    @Value("${rocketmq.consumer.group}")
    private String consumerGroup;

    @Value("${rocketmq.name.server}")
    private String nameSvr;

    @Value("${rocketmq.subscribe.topic}")
    private String topic;

    private String tag = "*";

    private Integer maxReconsumeTimes = 2;

    @Autowired
    private MessageListenerConcurrently messageListener;

    @Bean
    public DefaultMQPushConsumer defaultMQPushConsumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(nameSvr);
        consumer.subscribe(topic, "*");
        consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.registerMessageListener(messageListener);
        consumer.setMaxReconsumeTimes(maxReconsumeTimes);
        consumer.setMaxReconsumeTimes(3);
        consumer.start();
        log.info("Consumer Started.%n");
        return consumer;
    }
}
